#include "StdAfx.h"
#include "JsonStreamServer.h"
#include "json.h"
#include "MemoryManager.h"
#include "Types.h"
#include "Element.h"
#include "TimestampGenerator.h"
#include "DocumentIdentifierGenerator.h"
#include "QueueEntity.h"
#include "RelationSynopsis.h"
#include "LineageSynopsis.h"
#include "JsonStreamServer.h"
#include "JsonSchema.h"
#include "SchemaInterpreter.h"
#include "Operator.h"
#include "RowWindowOperator.h"
#include "JoinOperator.h"
#include "SelectionOperator.h"
#include "ProjectionOperator.h"
#include "RootOperator.h"
#include "RstreamOperator.h"
#include "IstreamOperator.h"
#include "DstreamOperator.h"
#include "LeafOperator.h"
#include "Scheduler.h"
#include "IOManager.h"
#include "SocketStreamInput.h"
#include "IStreamInput.h"
#include "RandomGeneratedStreamInput.h"
#include "ShowResultStreamOutput.h"
#include "IStreamOutput.h"
#include "QueryManager.h"
#include "QueryIntermediateRepresentation.h"
#include "JsonSchema.h"
#include "PlanManager.h"
#include "SchemaManager.h"
#include <windows.h>
JsonStreamServer* JsonStreamServer::jsonStreamServer = NULL;
JsonStreamServer::JsonStreamServer(void)
{
	
}


JsonStreamServer::~JsonStreamServer(void)
{

}
JsonStreamServer* JsonStreamServer:: getInstance(void)
{
	if(jsonStreamServer == NULL)
	{
		jsonStreamServer = new JsonStreamServer();
	}
	return jsonStreamServer;
}
void JsonStreamServer::beginSpecification()
{
	IOManager::getInstance();
	MemoryManager::getInstance();
	PlanManager::getInstance();
	QueryManager::getInstance();
	SchemaManager::getInstance();
}
void JsonStreamServer::endSpecification()
{
 
	initialScheduler();

}
void JsonStreamServer::initialScheduler()
{
	this->scheduler.reset(new Scheduler(PlanManager::getInstance()->getOperatorList()));
 
}


void JsonStreamServer::registerStream(boost::shared_ptr<IStreamInput> streamInput)
{
	 QueryManager::getInstance()->addStreamInput(streamInput);
}
void JsonStreamServer::registerQuery(boost::shared_ptr<QueryIntermediateRepresentation> queryIntermediateRepresentation,boost::shared_ptr<IStreamOutput> streamOutput)
{
	QueryManager::getInstance()->addStreamOutput(streamOutput);
	PlanManager::getInstance()->generateQueryPlan(queryIntermediateRepresentation);

}
//void JsonStreamServer::materialize (void)
//{
//	materialize1();
//	//materialize2();
//	//materialize3();
//}
//void JsonStreamServer::materialize1 (void) // test window,selection,projection,join operators
//{
//
//	//two leaf operator, two row-based window operator, one join operator, one selection operator, one projection operator, and istream operator
//	std::string schemaString1 = "{\"id\": \"stream1\",\"description\":\"a schema example1\",\"type\"\
//								:\"object\",\"properties\": {\
//								\"id\":{\"description\":\"total numbers\",\"type\": \"string\"}}}";
//	Document schemaDocument1 = fromjson(schemaString1);
//	boost::shared_ptr<JsonSchema> jsonSchema1(new JsonSchema(schemaDocument1));
//	boost::shared_ptr<IStreamInput> streamInput1 (new RandomGeneratedStreamInput(jsonSchema1));
//	boost::shared_ptr<LeafOperator> leafOperator1(new LeafOperator());// create leaf operator1
//	leafOperator1->setStreamInput(streamInput1);
//	this->operatorList.push_back(leafOperator1);
//
//
//	std::string schemaString2 = "{\"id\": \"stream1\",\"description\":\"a schema example1\",\"type\"\
//								:\"object\",\"properties\": {\
//								\"id\":{\"description\":\"total numbers\",\"type\": \"string\"},\
//								\"num\":{\"type\":\"number\"}}}";
//	Document schemaDocument2 = fromjson(schemaString2);
//	boost::shared_ptr<JsonSchema> jsonSchema2(new JsonSchema(schemaDocument2));
//	boost::shared_ptr<IStreamInput> streamInput2 (new RandomGeneratedStreamInput(jsonSchema2));
//	boost::shared_ptr<LeafOperator> leafOperator2(new LeafOperator());// create leaf operator2
//	leafOperator2->setStreamInput(streamInput2);
//	this->operatorList.push_back(leafOperator2);
//
//	int rowWindowSize1 = 3;
//	boost::shared_ptr<RowWindowOperator> rowWindowOperator1(new RowWindowOperator(rowWindowSize1));//create window operator1
//	this->operatorList.push_back(rowWindowOperator1);
//
//	int rowWindowSize2 = 3;
//	boost::shared_ptr<RowWindowOperator> rowWindowOperator2(new RowWindowOperator(rowWindowSize2));//create window operator2
//	this->operatorList.push_back(rowWindowOperator2);
//
//	boost::shared_ptr<JoinOperator> joinOperator(new JoinOperator());//create join operator
//	this->operatorList.push_back(joinOperator);
//	joinOperator->setLeftJoinAttribute("id");
//	joinOperator->setRightJoinAttribute("id");
//
//	boost::shared_ptr<SelectionOperator> selectionOperator(new SelectionOperator());//create selection operator
//	this->operatorList.push_back(selectionOperator);
//
//	std::set<std::string>projectionAttributeSet;
//	projectionAttributeSet.insert("id");
//	boost::shared_ptr<ProjectionOperator> projectionOperator(new ProjectionOperator(projectionAttributeSet));//create projection operator
//	this->operatorList.push_back(projectionOperator);
//
//	boost::shared_ptr<IstreamOperator> istreamOperator(new IstreamOperator());
//	this->operatorList.push_back(istreamOperator);
//
//	//show result stream output
//	//boost::shared_ptr<RootOperator> rootOperator(new RootOperator());//create root operator
//	//this->operatorList.push_back(rootOperator);
//	//boost::shared_ptr<IStreamOutput> streamOutput(new ShowResultStreamOutput());
//	//rootOperator->setStreamOutput(streamOutput);
//
//	//socket stream output
//	std::string ip = "localhost";
//	std::string port = "8080";
//	boost::shared_ptr<RootOperator> rootOperator(new RootOperator());//create root operator
//	this->operatorList.push_back(rootOperator);
//	boost::shared_ptr<IStreamOutput> streamOutput(new SocketStreamOutput(ip,port));
//	rootOperator->setStreamOutput(streamOutput);
//
//
//	//create queues
//	boost::shared_ptr<QueueEntity>queueEntity1(new QueueEntity());
//	boost::shared_ptr<QueueEntity>queueEntity2(new QueueEntity());
//	boost::shared_ptr<QueueEntity>queueEntity3(new QueueEntity());
//	boost::shared_ptr<QueueEntity>queueEntity4(new QueueEntity());
//	boost::shared_ptr<QueueEntity>queueEntity5(new QueueEntity());
//	boost::shared_ptr<QueueEntity>queueEntity6(new QueueEntity());
//	boost::shared_ptr<QueueEntity>queueEntity7(new QueueEntity());
//	boost::shared_ptr<QueueEntity>queueEntity8(new QueueEntity());
//
//	leafOperator1->setOutputQueue(queueEntity1);
//	rowWindowOperator1->setInputQueue(queueEntity1);
//	leafOperator2->setOutputQueue(queueEntity2);
//	rowWindowOperator2->setInputQueue(queueEntity2);
//
//	rowWindowOperator1->setOutputQueue(queueEntity3);
//	joinOperator->setLeftInputQueue(queueEntity3);
//
//	rowWindowOperator2->setOutputQueue(queueEntity4);
//	joinOperator->setRightInputQueue(queueEntity4);
//
//	joinOperator->setOutputQueue(queueEntity5);
//	selectionOperator->setInputQueue(queueEntity5);
//
//	selectionOperator->setOutputQueue(queueEntity6);
//	projectionOperator->setInputQueue(queueEntity6);
//
//	projectionOperator->setOutputQueue(queueEntity7);
//	istreamOperator->setInputQueue(queueEntity7);
//
//	istreamOperator->setOutputQueue(queueEntity8);
//	rootOperator->setInputQueue(queueEntity8);
//#ifdef DEBUG
//	leafOperator1->setId("leafOperator1");
//	leafOperator2->setId("leafOperator2");
//	rowWindowOperator1->setId("rowWindowOperator1");
//	rowWindowOperator2->setId("rowWindowOperator2");
//	joinOperator->setId("joinOperator");
//	selectionOperator->setId("selectionOperator");
//	projectionOperator->setId("projectionOperator");
//	istreamOperator->setId("istreamOperator");
//	rootOperator->setId("rootOperator");
//	queueEntity1->setId("queueEntity1");
//	queueEntity2->setId("queueEntity2");
//	queueEntity3->setId("queueEntity3");
//	queueEntity4->setId("queueEntity4");
//	queueEntity5->setId("queueEntity5");
//	queueEntity6->setId("queueEntity6");
//	queueEntity7->setId("queueEntity7");
//	queueEntity8->setId("queueEntity8");
//#endif
//}
//void JsonStreamServer::materialize2 (void) // test istream,dstream,rstream operators
//{
//	// leafoperator -> rowwindowoperator -> rstream/istream/dstream operator -> root operator
//	std::string schemaString = "{\"id\": \"stream1\",\"description\":\"a schema example1\",\"type\"\
//							   :\"object\",\"properties\": {\
//							   \"id\":{\"description\":\"total numbers\",\"type\": \"string\"}}}";
//	Document schemaDocument = fromjson(schemaString);
//	boost::shared_ptr<JsonSchema> jsonSchema(new JsonSchema(schemaDocument));
//	boost::shared_ptr<IStreamInput> streamInput (new RandomGeneratedStreamInput(jsonSchema));
//	boost::shared_ptr<LeafOperator> leafOperator(new LeafOperator());// create leaf operator1
//	leafOperator->setStreamInput(streamInput);
//	this->operatorList.push_back(leafOperator);
//
//	int rowWindowSize = 3;
//	boost::shared_ptr<RowWindowOperator> rowWindowOperator(new RowWindowOperator(rowWindowSize));//create window operator
//	this->operatorList.push_back(rowWindowOperator);
//
//	//test rstream
//	//boost::shared_ptr<RstreamOperator> relation_to_stream(new RstreamOperator());
//
//	//test istream
//	//boost::shared_ptr<IstreamOperator> relation_to_stream(new IstreamOperator());
//
//	//test dstream
//	boost::shared_ptr<DstreamOperator> relation_to_stream(new DstreamOperator());
//
//	this->operatorList.push_back(relation_to_stream);
//
//	boost::shared_ptr<RootOperator> rootOperator(new RootOperator());//create root operator
//	this->operatorList.push_back(rootOperator);
//	boost::shared_ptr<IStreamOutput> streamOutput(new ShowResultStreamOutput());
//	rootOperator->setStreamOutput(streamOutput);
//
//	//create queues
//	boost::shared_ptr<QueueEntity>queueEntity1(new QueueEntity());
//	boost::shared_ptr<QueueEntity>queueEntity2(new QueueEntity());
//	boost::shared_ptr<QueueEntity>queueEntity3(new QueueEntity());
//
//
//	leafOperator->setOutputQueue(queueEntity1);
//	rowWindowOperator->setInputQueue(queueEntity1);
//
//	rowWindowOperator->setOutputQueue(queueEntity2);
//	relation_to_stream->setInputQueue(queueEntity2);
//
//	relation_to_stream->setOutputQueue(queueEntity3);
//	rootOperator->setInputQueue(queueEntity3);
//#ifdef DEBUG
//	leafOperator->setId("leafOperator");
//	relation_to_stream->setId("relation_to_stream_Operator");
//	rootOperator->setId("rootOperator");
//	queueEntity1->setId("queueEntity1");
//	queueEntity2->setId("queueEntity2");
//	queueEntity3->setId("queueEntity3");
//
//#endif
//}
//
//void JsonStreamServer::materialize3 (void) // test leaf operators
//{
//	// leafoperator -> rowwindowoperator ->  root operator
//	std::string schemaString = "{\"id\": \"stream1\",\"description\":\"a schema example1\",\"type\"\
//							   :\"object\",\"properties\": {\
//							   \"id\":{\"description\":\"total numbers\",\"type\": \"string\"}}}";
//	Document schemaDocument = fromjson(schemaString);
//	boost::shared_ptr<JsonSchema> jsonSchema(new JsonSchema(schemaDocument));
//	boost::shared_ptr<LeafOperator>leafOperator(new LeafOperator());
//	this->operatorList.push_back(leafOperator);
//	std::string ip = "localhost";
//	std::string port = "8080";
//	boost::shared_ptr<IStreamInput> streamInput(new SocketStreamInput(ip,port,jsonSchema));
//	leafOperator->setStreamInput(streamInput);
//
//
//	int rowWindowSize = 3;
//	boost::shared_ptr<RowWindowOperator> rowWindowOperator(new RowWindowOperator(rowWindowSize));//create window operator
//	this->operatorList.push_back(rowWindowOperator);
//
//
//	boost::shared_ptr<RootOperator> rootOperator(new RootOperator());//create root operator
//	this->operatorList.push_back(rootOperator);
//	boost::shared_ptr<IStreamOutput> streamOutput(new ShowResultStreamOutput());
//	rootOperator->setStreamOutput(streamOutput);
//
//	//create queues
//	boost::shared_ptr<QueueEntity>queueEntity1(new QueueEntity());
//	boost::shared_ptr<QueueEntity>queueEntity2(new QueueEntity());
//
//
//
//	leafOperator->setOutputQueue(queueEntity1);
//	rowWindowOperator->setInputQueue(queueEntity1);
//
//	rowWindowOperator->setOutputQueue(queueEntity2);
//	rootOperator->setInputQueue(queueEntity2);
//#ifdef DEBUG
//	leafOperator->setId("leafOperator");
//	rowWindowOperator->setId("rowWindowOperator");
//	rootOperator->setId("rootOperator");
//	queueEntity1->setId("queueEntity1");
//	queueEntity2->setId("queueEntity2");
//
//
//#endif
//}
void JsonStreamServer::execute()
{
	//for(int i = 0 ; i<24 ;i++)
	while(1)
	{
		IOManager::getInstance()->executeOnce();
		boost::shared_ptr<Operator> op;
		int executionTimes;
		this->scheduler->getNextOperatorToExecute(op,executionTimes);
		op->execution(executionTimes);
		//Sleep(2000);
	}
}